package indi.zhifa.recipe.bailan.framework.queue.config;

import indi.zhifa.recipe.bailan.framework.queue.bean.RedisConsumerMap;
import indi.zhifa.recipe.bailan.framework.queue.handler.RedisConsumer;
import indi.zhifa.recipe.bailan.framework.queue.property.ConsumerConfig;
import indi.zhifa.recipe.bailan.framework.queue.property.RedisQueueProperty;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.CollectionUtils;

import java.time.Duration;
import java.util.Iterator;
import java.util.List;

@RequiredArgsConstructor
@Configuration
@Slf4j
public class RedisQueueConfig {

    private final RedisConnectionFactory mRedisConnectionFactory;
    private final ThreadPoolTaskExecutor mThreadPoolTaskExecutor;
    private final RedisQueueProperty redisQueueProperty;
    private final RedisConsumerMap mRedisConsumerMap;
    private final RedisTemplate<String,Object> mRedisTemplate;
    StreamMessageListenerContainer<String, MapRecord<String,Object,byte[]>>[] streamMessageListenerContainers;

    @Bean
    public StreamMessageListenerContainer<String, MapRecord<String,Object,byte[]>>[] redisMesListeners() {

        List<ConsumerConfig> consumerPairList = redisQueueProperty.getConsumerPairs();
        if(CollectionUtils.isEmpty(consumerPairList)){
            return null;
        }
        streamMessageListenerContainers = new StreamMessageListenerContainer[consumerPairList.size()];
        StreamOperations<String,Object,byte[]> streamOperations = mRedisTemplate.opsForStream();
        for(int i=0;i<consumerPairList.size();i++){
            ConsumerConfig consumerConfig = consumerPairList.get(i);
            RedisConsumer redisConsumer = mRedisConsumerMap.getConsumer(consumerConfig.getName());
            boolean groupCreated = false;
            if(mRedisTemplate.hasKey(consumerConfig.getStream())){
                StreamInfo.XInfoGroups groups = streamOperations.groups(consumerConfig.getStream());
                if(null != groups){
                    Iterator<StreamInfo.XInfoGroup> it = groups.iterator();
                    while (it.hasNext()){
                        StreamInfo.XInfoGroup group = it.next();
                        if(group.groupName().equals(consumerConfig.getGroup())){
                            groupCreated = true;
                            break;
                        }
                    }
                }
            }

            if(!groupCreated){
                streamOperations.createGroup(consumerConfig.getStream(),consumerConfig.getGroup());
            }

            StreamMessageListenerContainer<String, MapRecord<String,Object,byte[]>> container =
                    streamContainer(consumerConfig.getStream(), consumerConfig.getGroup(), consumerConfig.getName(),
                            consumerConfig.getBitchSize(),consumerConfig.getTimeOut(),
                            redisConsumer);
            streamMessageListenerContainers[i] = container;
            container.start();
        }

        return streamMessageListenerContainers;
    }

    private StreamMessageListenerContainer<String, MapRecord<String,Object,byte[]>> streamContainer(String pStream, String pGroup, String pConsumer,
                                                                                                 int pBatchSize,long pTimeOut,
                                                                                                 RedisConsumer pRedisConsumer){

        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String,Object,byte[]>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                .pollTimeout(Duration.ofMillis(pTimeOut))
                .batchSize(pBatchSize)
                .executor(mThreadPoolTaskExecutor)
                 .keySerializer(RedisSerializer.string())
                 .hashKeySerializer(RedisSerializer.string())
                .hashValueSerializer(RedisSerializer.byteArray())
                .build();

        StreamMessageListenerContainer<String, MapRecord<String,Object,byte[]>> container = StreamMessageListenerContainer
                .create(mRedisConnectionFactory, options);

        //指定消费最新的消息
        StreamOffset<String> offset = StreamOffset.create(pStream, ReadOffset.lastConsumed());

        //创建消费者
        Consumer consumer = Consumer.from(pGroup, pConsumer);

        StreamMessageListenerContainer.StreamReadRequest<String> streamReadRequest = StreamMessageListenerContainer.StreamReadRequest.builder(offset)
                .errorHandler((error)->{
                    if(null != pRedisConsumer.getErrMsgHandler()){
                        pRedisConsumer.getErrMsgHandler().handleError(error);
                    }else{
                        log.error(error.getMessage());
                    }
                })
                .cancelOnError(e -> false)
                .consumer(consumer)
                //关闭自动ack确认
                .autoAcknowledge(false)
                .build();
        //指定消费者对象
        container.register(streamReadRequest, pRedisConsumer.getRedisMsgConsumerHandler());

        return container;
    }

}
